package com.netposa.kafka;

import org.apache.kafka.clients.producer.KafkaProducer;

import java.util.Properties;

public class ProducerFactory {

    private static ProducerFactory instance = null;
    private static Properties props = null;
    private KafkaProducer<String, String> kafkaProducer;

    static {
        props = new Properties();
        props.put("bootstrap.servers", "192.168.43.122:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    }


    public KafkaProducer getProducer() {
        if (kafkaProducer == null) {
            kafkaProducer = new KafkaProducer<>(props);
        }
        return kafkaProducer;
    }


    //单例模式，双重检查
    public static ProducerFactory getInstance() {
        if (instance == null) {
            synchronized (ProducerFactory.class) {
                if (instance == null) {
                    instance = new ProducerFactory();
                }
            }
        }
        return instance;
    }

    //单例模式，静态内部类
    private static class ProducerFactoryInner {
        public static final ProducerFactory instance = new ProducerFactory();
    }

    public static ProducerFactory getInstance2() {
        return ProducerFactoryInner.instance;
    }

}
